package org.code.util;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.code.common.Constant;

/**
 * @author xiao
 * <p>
 * RocketMQ工具类
 * 封装消息消费者、消息生产者
 */
public class RocketMQUtil {
    /**
     * nameserver地址
     */
    private static final String NAME_SERVER = Constant.NAME_SERVER;
    /**
     * 默认消息生产者组
     */
    private static final String DEFAULT_PRODUCER_GROUP = "defaultProducerGroup";
    /**
     * 默认消息消费者组
     */
    public static final String DEFAULT_CONSUMER_GROUP = "defaultConsumerGroup";

    /**
     * 获取消息生产者
     *
     * @param producerGroup  消息生产者组
     * @param nameServerAddr nameserver地址
     * @return 消息生产者
     */
    public static DefaultMQProducer getProducer(final String producerGroup, final String nameServerAddr) throws MQClientException {
        DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr(nameServerAddr);
        producer.start();
        return producer;
    }

    /**
     * 获取消息生产者
     *
     * @param producerGroup 消息生产者组
     * @return 消息生产者
     */
    public static DefaultMQProducer getProducer(final String producerGroup) throws MQClientException {
        return getProducer(producerGroup, NAME_SERVER);
    }

    /**
     * 获取消息生产者
     *
     * @return 消息生产者
     */
    public static DefaultMQProducer getProducer() throws MQClientException {
        return getProducer(DEFAULT_PRODUCER_GROUP);
    }

    /**
     * 获取消息消费者
     *
     * @param consumerGroup  消息消费者组
     * @param nameServerAddr nameserver地址
     * @return 消息消费者
     */
    public static DefaultMQPushConsumer getConsumer(final String consumerGroup, final String nameServerAddr) {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(nameServerAddr);
        return consumer;
    }

    /**
     * 获取消息消费者
     *
     * @param consumerGroup 消息消费者组
     * @return 消息消费者
     */
    public static DefaultMQPushConsumer getConsumer(final String consumerGroup) {
        return getConsumer(consumerGroup, NAME_SERVER);
    }

    /**
     * 获取消息消费者
     *
     * @return 消息消费者
     */
    public static DefaultMQPushConsumer getConsumer() {
        return getConsumer(DEFAULT_CONSUMER_GROUP);
    }

    /**
     * 关闭消息生产者、消息消费者
     *
     * @param object 消息生产者、消息消费者
     */
    public static void shutdown(Object object) {
        if (object instanceof DefaultMQProducer) {
            ((DefaultMQProducer) object).shutdown();
        } else if (object instanceof DefaultMQPushConsumer) {
            ((DefaultMQPushConsumer) object).shutdown();
        } else {
            throw new IllegalArgumentException("object is not DefaultMQProducer or DefaultMQPushConsumer");
        }
    }
}
